package com.yunhe.transfer.service.data.impl;

import com.google.protobuf.ByteString;
import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.common.constant.DpuConsts;
import com.yunhe.common.constant.MqttConsts;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.transfer.proto.EigProto;
import com.yunhe.transfer.service.data.DpuFileService;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

/**
 * DPU采集和控制配置文件下发、查看service实现
 *
 * @author duanxiaoyong
 */
@Service
public class DpuFileServiceImpl implements DpuFileService {

    @Autowired
    private AuthorityClient authorityClient;
    @Autowired
    private MqttClient mqttClient;

    private static final String INFILL_AOE = "AOE";
    private static final String INFILL_POINT = "PF";
    private static final String INFILL_CHANNEL = "TF";

    private static final Logger logger = LoggerFactory.getLogger(DpuFileServiceImpl.class);

    @Override
    public Boolean send(String type, Optional<Long> stationId, Optional<String> beeId, MultipartFile file)
            throws Exception {
        if (!stationId.isPresent() && !beeId.isPresent()) {
            throw new ArgumentErrorException("请指定电站或beeId");
        }
        List<String> topics = new ArrayList<>();
        if (DpuConsts.FILE_TYPE.Channel.value().equals(type)) {
            getTopics(topics, stationId, beeId, MqttConsts.TOPIC.DPU_TpFile_Client);
        } else if (DpuConsts.FILE_TYPE.Point.value().equals(type)) {
            getTopics(topics, stationId, beeId, MqttConsts.TOPIC.DPU_PtFile_Client);
        } else if (DpuConsts.FILE_TYPE.AOE.value().equals(type)) {
            getTopics(topics, stationId, beeId, MqttConsts.TOPIC.DPU_AoeFile_Client);
        }
        if (topics.size() > 0 && file.getOriginalFilename() != null) {
            // 获取文件byte数组
            byte[] bytes = file.getBytes();
            // 将文件内容转成com.google.protobuf.ByteString类型
            ByteString content = ByteString.copyFrom(bytes);
            // 生成pbFile文件
            EigProto.pbFile pbFile = EigProto.pbFile.newBuilder().
                    setFileName(file.getOriginalFilename()).
                    setFileContent(content).
                    build();
            // 发布MQTT主题消息
            for (String topic : topics) {
                mqttClient.publish(topic, pbFile.toByteArray(), 1, false);
            }
        }
        return true;
    }

    @Override
    public void export(String type,
                       String beeId,
                       String fileName,
                       HttpServletResponse response)
            throws Exception {
        String fileType = getFileType(type);
        if (fileType == null) {
            return;
        }
        if (fileName != null) {
            // 先订阅主题
            String lock = UUID.randomUUID().toString();
            String topic = MqttConsts.TOPIC.DPU_File_Server.value(beeId + "-" + fileType + "-" + fileName);
            try {
                mqttClient.subscribe(topic, 1, (tp, msg) -> {
                    synchronized (lock) {
                        byte[] bytes = fileName.getBytes(StandardCharsets.UTF_8);
                        byte[] payload = msg.getPayload();
                        if (payload != null && payload.length > bytes.length + 9) {
                            response.reset();
                            response.setContentType("application/csv; charset=UTF-8");
                            response.setHeader("Content-Disposition", "attachment; filename=" + fileName);
                            ServletOutputStream outputStream = response.getOutputStream();
                            outputStream.write(Arrays.copyOfRange(payload, bytes.length + 9, payload.length));
                            outputStream.close();
                        }
                        lock.notify();
                    }
                });
            } catch (Exception e) {
                logger.error("获取DPU文件出错，主题：{}", topic, e);
            }
            // 然后重置，DPU发送最新消息
            mqttClient.publish(MqttConsts.TOPIC.DPU_Reset_Client.value(beeId), new byte[0], 1, false);
            synchronized (lock) {
                lock.wait(5000);
            }
        } else {
            String topic = MqttConsts.TOPIC.DPU_File_Server.value("#");
            String prefix = MqttConsts.TOPIC.DPU_File_Server.value("");
            Map<String, byte[]> fileMap = new HashMap<>();
            try {
                mqttClient.subscribe(topic, 1, (tp, msg) -> {
                    if (tp.startsWith(MqttConsts.TOPIC.DPU_File_Server.value(beeId + "-" + fileType))) {
                        String fileUrl = tp.replace(prefix, "");
                        String fName = fileUrl.replace(beeId + "-" + fileType + "-", "");
                        byte[] bytes = fName.getBytes(StandardCharsets.UTF_8);
                        byte[] payload = msg.getPayload();
                        if (payload != null && payload.length > bytes.length + 9) {
                            fileMap.put(fName, Arrays.copyOfRange(payload, bytes.length + 9, payload.length));
                        }
                    }
                });
            } catch (Exception e) {
                logger.error("获取DPU文件出错，主题：{}", topic, e);
            }
            // 然后重置，DPU发送最新消息
            mqttClient.publish(MqttConsts.TOPIC.DPU_Reset_Client.value(beeId), new byte[0], 1, false);
            // 等待5s，接收结果
            int delay = 1000;
            int waitTime = 0;
            int maxWaitTime = 5000;
            int size = fileMap.size();
            while (true) {
                Thread.sleep(delay);
                waitTime += delay;
                if (fileMap.size() == size || waitTime >= maxWaitTime) {
                    // 当文件数不在增加或者等待时间大于5s，结束等待
                    break;
                }
                size = fileMap.size();
            }
            // 导出zip文件
            response.reset();
            response.setContentType("application/x-download; charset=UTF-8");
            response.setHeader("Content-Disposition", "attachment; filename=" + beeId + ".zip");
            OutputStream outputStream = response.getOutputStream();
            ZipOutputStream zipOutputStream = new ZipOutputStream(new BufferedOutputStream(outputStream));
            zipOutputStream.setMethod(ZipOutputStream.DEFLATED);
            for (Map.Entry<String, byte[]> entry : fileMap.entrySet()) {
                zipOutputStream.putNextEntry(new ZipEntry(entry.getKey()));
                zipOutputStream.write(entry.getValue());
                zipOutputStream.closeEntry();
            }
            zipOutputStream.close();
            outputStream.close();
        }
    }

    private String getFileType(String type) {
        if (DpuConsts.FILE_TYPE.Channel.value().equals(type)) {
            return INFILL_CHANNEL;
        } else if (DpuConsts.FILE_TYPE.Point.value().equals(type)) {
            return INFILL_POINT;
        } else if (DpuConsts.FILE_TYPE.AOE.value().equals(type)) {
            return INFILL_AOE;
        } else {
            return null;
        }
    }

    private void getTopics(List<String> topics,
                           Optional<Long> stationId,
                           Optional<String> beeId,
                           MqttConsts.TOPIC topic) {
        if (stationId.isPresent() && !beeId.isPresent()) {
            Station station = authorityClient.getStationById(stationId.get()).getResults();
            List<String> beeIds = station.getBeeIds();
            if (beeIds != null && beeIds.size() > 0) {
                beeIds.forEach(bId -> topics.add(topic.value(bId)));
            }
        } else {
            beeId.ifPresent(bId -> topics.add(topic.value(bId)));
        }
    }
}
